package com.test

import org.junit.jupiter.api.DisplayName
import org.junit.jupiter.api.Test
import org.slf4j.LoggerFactory
import reactor.core.publisher.Flux
import reactor.core.publisher.Mono
import reactor.core.publisher.SynchronousSink
import reactor.core.publisher.onErrorReturn
import java.lang.RuntimeException
import java.time.Duration
import java.util.concurrent.Callable
import java.util.function.BiFunction
import java.util.function.Function
import kotlin.math.sin

@DisplayName("ReactorTestKt 测试")
class ReactorTestKt {
    val log = LoggerFactory.getLogger(this.javaClass);
    @Test
    fun onErrorReturn() {
        Flux
                .interval(Duration.ofMillis(1000))
                .map {
                    if (it == 3L || it == 4L || it == 5L || it == 6L) {
                        throw RuntimeException("mock exception")
                    }
                    it.toString() + "0"
                }
                .doOnError{ e -> log.error("error type:{}, error message:{}", e.javaClass, e.message)}
                .onErrorResume{_ -> Flux.range(1,3).map { it.toString() }}
                .log()
                .subscribe();
        Thread.sleep(10000);
    }

    @Test
    internal fun generateTest() {
        val objectFlux : Flux<Any> = Flux
                .generate(
                        { 0 },
                        { i: Int, sink: SynchronousSink<Any> -> sink.next(i * i)
                            if( i == 5) sink.complete()
                            i+1
                        },
                        { log.info(it.toString())}
                )
                .log();
        objectFlux.subscribe();
    }

    @Test
    internal fun createTest() {
        val objectMono : Mono<Any> = Mono
                .create({
                    var list : Array<Int> = arrayOf(1, 2, 3, 4, 5)
                    it.success(list)
                })
    }
}